package com.fjh.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fjh.config.ProductThreadPool;
import com.fjh.constant.QueueConstant;
import com.fjh.dao.ProEsDao;
import com.fjh.domain.Prod;
import com.fjh.es.ProdEs;
import com.fjh.model.EsChange;
import com.fjh.service.ImportService;
import com.fjh.service.ProdService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import cn.hutool.core.bean.BeanUtil;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

@Service
@Slf4j
public class ImportServiceImpl implements ImportService,CommandLineRunner{

    @Autowired
    private ProEsDao proEsDao;

    @Autowired
    private ProdService prodService;

    @Value("${esimport.size}")
    private Integer size;

    private Date t1;

    /**
     * 全量导入
     */
    @Override
    public void importAll() {
    log.info("全量导入开始");
    //查询总条数
        Integer count = prodService.totalCount(null,null);
        if (count == null || count == 0){
            log.info("count=0  没有商品导入");
            return;
        }
        //计算总页数
        int page = count % size == 0 ? count / size : ((count / size) + 1);
        //创建一个线程工具类
        CountDownLatch countDownLatch = new CountDownLatch(page);
        for (int i = 0; i < page; i++) {
            //查询数据后导入es
            final int a = i;
            ProductThreadPool.poolExecutor.execute(()->{
                importToEs(a,size,null,null);
                countDownLatch.countDown();
            });
        }
        log.info("全量导入完成");
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        // 从这里开始 记录一个时候点  这个时间点是提交线程任务的时间点
        t1 = new Date();
        System.out.println(t1);
    }

    /**
     * 导入es
     * @param i
     * @param size
     * @param t1
     * @param t2
     */
    private void importToEs(int i, Integer size, Date t1, Date t2) {
        Page<Prod> page = new Page<>(i, size, false);
        List<Prod> prods = prodService.findProdToEs(page,t1,t2);
        //转换成prodEs对象
        ArrayList<ProdEs> prodEs = new ArrayList<>(prods.size());
        prods.forEach(prod -> {
            ProdEs prodEs1 = new ProdEs();
            //属性拷贝
            BeanUtil.copyProperties(prod,prodEs1,true);
            prodEs.add(prodEs1);
        });
        //再插入es
        proEsDao.saveAll(prodEs);
        System.out.println("导入完成的：" + new Date());
    }

    /**
     * 增量导入   查询在某个时间段内修改过的数据  10.00  - 10.30
     * 商家上新商品等
     * 2分钟间隔
     * initialDelay 项目启动以后多少秒执行第一次
     * fixedRate  第一次执行以后 后面的执行周期
     */
    @Override
    @Scheduled(initialDelay = 60*1000, fixedRate = 60*1000)
    public void updateImport() {
        log.info("增量导入开始");
        Date t2 = new Date();
        System.out.println(t1);
        System.out.println(t2);
        //查询总条数
        Integer count = prodService.totalCount(t1, t2);
        if (count == null || count == 0){
            t1 = t2;
            log.info("count = 0 没有商品导入");
            return;
        }
        //计算总页数
        int page = count % size == 0 ? count/size : ((count/size) +1);
        //线程辅助工具类  countDown  CyclicBarrier  Semaphone
        CountDownLatch downLatch = new CountDownLatch(page);
        for (int i = 1; i < page; i++) {
            final int a = i;
            //查询数据后导入es
            ProductThreadPool.poolExecutor.execute(()->{
                importToEs(a,size,t1,t2);
                downLatch.countDown();
            });
        }
        log.info("增量导入完成");
        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        t1 = t2;
    }

    /**
     * 快速导入
     * 购买商品，紧急情况
     * 快速同步es
     * 用户在order-service 同步扣减库存 下单减库存 mysql  es
     * 下单需要做很多事情  我们需要扣减es的库存 不能用同步方式 我们需要异步的方式
     * 使用mq做异步 解耦合
     * 当用户下单 以后 把商品的id 和 count 放入mq
     * 我们在product-service做监听
     */
    @Override
    public void qiuckImport() {
    }

    /**
     * 快速导入
     * 购买商品，紧急情况
     * 快速同步es
     * 用户在order-service 同步扣减库存 下单减库存 mysql  es
     * 下单需要做很多事情  我们需要扣减es的库存 不能用同步方式 我们需要异步的方式
     * 使用mq做异步 解耦合
     * 当用户下单 以后 把商品的id 和 count 放入mq
     * 我们在product-service做监听
     * @param message
     * @param channel
     */
    @RabbitListener(queues = QueueConstant.PROD_CHANGE_QUEUE,concurrency = "3-5")
    public void handleEsMsg(Message message, Channel channel){
        String s = new String(message.getBody());
        List<EsChange> esChanges = JSON.parseArray(s, EsChange.class);
        //先拿到商品的ids
        List<Long> prodIds = esChanges.stream()
                .map(EsChange::getProdId)
                .collect(Collectors.toList());
        //查询es  拿到对应的集合
        Iterable<ProdEs> prodEsIterable = proEsDao.findAllById(prodIds);
        //循环集合找到对应的商品，修改对应的库存
        prodEsIterable.forEach(prodEs -> {
            //找到对应的数据
            EsChange change = esChanges.stream()
                    .filter(esChange -> esChange.getProdId().equals(prodIds))
                    .collect(Collectors.toList()).get(0);
            // 修改库存
            long finalStock = prodEs.getTotalStocks() + change.getCount();
            if (finalStock < 0) {
                log.error("es中{}的库存不足", prodEs.getProdId());
                throw new IllegalArgumentException("es中" + prodEs.getProdId() + "的库存不足");
            }
            prodEs.setTotalStocks(finalStock);
            // 设置销量
            prodEs.setSoldNum(prodEs.getSoldNum() - change.getCount());
        });
        // 统一saveAll 如果es中有 就是更新
        proEsDao.saveAll(prodEsIterable);
        // 手动签收
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 启动就会执行
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        importAll();
    }
}
